/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月22日 下午3:23:26
*/
package cn.com.guage.flink.window;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction.Context;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

/**
 * 测试ProcessWinFunction
 *
 * @author dajiangtai
 * @create 2019-06-11-18:37
 */
public class TestProcessWinFunctionOnWindow {

	public static void main(String[] args) throws Exception {
		// 获取执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// 读取数据
		DataStream<Tuple3<String, String, Long>> input = env.fromElements(ENGLISH);
		DataStream<Tuple3<String,String,Long>> keybyStream = input.keyBy(new KeySelector<Tuple3<String,String,Long>,Tuple2<String,Long>>(){

			@Override
			public Tuple2<String, Long> getKey(Tuple3<String, String, Long> value) throws Exception {
				return Tuple2.of(value.f0, value.f2);
			}
		});
		keybyStream.print();
		 
		
		//process(new MyProcessWindowFunction());
		// 求各班级英语成绩平均分
		DataStream<Double> avgScore = input.keyBy(0).countWindow(2).process(new MyProcessWindowFunction());
		avgScore.print();
		env.execute("TestProcessWinFunctionOnWindow");

	}

	public static final Tuple3[] ENGLISH = new Tuple3[] { Tuple3.of("class1", "张三", 100L),
			Tuple3.of("class1", "李四", 78L), Tuple3.of("class1", "王五", 99L), Tuple3.of("class2", "赵六", 81L),
			Tuple3.of("class2", "小七", 59L), Tuple3.of("class2", "小八", 97L), };
}